package test.sunyu.storm.demo.demo4;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import sunyu.kit.common.LogKit;

/**
 * @author 孙宇
 */
public class TopologyMain {

    private static final Logger logger = LogKit.getLogger();

    public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("myKafkaSpout", new MyKafkaSpout(), 2);
        builder.setBolt("myBolt1", new MyBolt1()).localOrShuffleGrouping("myKafkaSpout");

        Config conf = new Config();
        conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
        conf.put("kafkaServers", "127.0.0.1:9092");
        conf.put("kafkaTopic", "test.topic");
        conf.put("kafkaGroup", "test.group");

        if (args != null && args.length > 0) {
            logger.info("正在提交拓扑到集群处理");
            conf.setNumWorkers(3);
            try {
                StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
            } catch (AlreadyAliveException e) {
                e.printStackTrace();
            } catch (InvalidTopologyException e) {
                e.printStackTrace();
            } catch (AuthorizationException e) {
                e.printStackTrace();
            }
        } else {
            logger.info("正在提交拓扑到本地处理");
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("myTopologie", conf, builder.createTopology());
            Utils.sleep(Integer.MAX_VALUE);
            cluster.killTopology("myTopologie");
            cluster.shutdown();
        }
    }

}
